package messaging;

import java.util.HashMap;
import java.util.Hashtable;
import java.util.Map;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.Topic;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class SubscriberGateway {

	// One connection is established per client - all Gateways use the same
	// connection but different sessions
	private static Connection connection;
	protected Session session;
	protected Destination destination;

	// Mapping between name and active consumers (durable subscribers) for this
	// given session
	protected Map<String, MessageConsumer> activeConsumers;

	// Save the number of instances
	private volatile static int count = 0;

	// Save the name if you wish to later identify the Gateway
	private String destinationName;

	// Logger
	// private static Logger logger =
	// LogManager.getLogger(ObserverGateway.class.getName());

	/**
	 * After creation of the gateway the connection and session are already
	 * open. Resources must be released by calling close() when the gateway is
	 * no longer needed.
	 * 
	 * @param connectionFactoryName Defines the name of the connection factory which is looked up using JNDI.
	 * @param destinationName Defines the name for a destination (JMS queue or Topic) which is going to be observed.
	 * @throws NamingException Thrown when JNDI is unable to find the connectionFactory or the initial context.
	 * @throws JMSException Thrown when the JMS provider is unavailable i.e. connection or session cannot be established or the destination does not exist.
	 */
	public SubscriberGateway(String connectionFactoryName,
			String destinationName) throws NamingException, JMSException {
		activeConsumers = new HashMap<String, MessageConsumer>();
		synchronized (SubscriberGateway.class) {
			SubscriberGateway.count++;
		}

		Hashtable<String, String> environment = new Hashtable<String, String>();
		// Use default settings if config is unavailable
		environment.put(Context.INITIAL_CONTEXT_FACTORY,
				"org.exolab.jms.jndi.InitialContextFactory");
		environment.put(Context.PROVIDER_URL, "tcp://localhost:3035");
		environment.put(Context.SECURITY_PRINCIPAL, "admin");
		environment.put(Context.SECURITY_CREDENTIALS, "openjms");

		// get the InitialContext
		Context context = new InitialContext(environment);

		// Set up the connection
		synchronized (SubscriberGateway.class) {
			if (connection == null) {
				ConnectionFactory connectionFactory = (ConnectionFactory) context
						.lookup(connectionFactoryName);
				connection = connectionFactory.createConnection();

				connection.start();

				// logger.debug("First connection established.");
			}
		}

		// Set up a new session using the connection
		session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

		// Set up the message consumer using a destination
		destination = (Destination) context.lookup(destinationName);

		// Get rid of the context
		context.close();
	}

	/**
	 * Register a MessageListener which gets called whenever a new message is available at the destination.
	 * 
	 * @param subscriber Defines the MessageListener which gets called when a message arrives at a destination.
	 * @throws JMSException Throws an exception when consumer cannot be created due to problems with session or connection.
	 */
	public void registerSubscriber(MessageListener subscriber)
			throws JMSException {
		// Registering a consumer with a session is easy and does not need to be
		// accounted for when closing the session and/or connection
		session.createConsumer(destination).setMessageListener(subscriber);
	}

	/**
	 * Creates or resumes a durable subscription to a JMSTopic. It is necessary
	 * that the destination for the gateway is a topic to create a durable
	 * subscription. There can be only one durable subscriber active at a given
	 * time. The durable subscription holds on to any messages arriving at the
	 * topic while the subscriber is inactive. Once created, the JMS provider
	 * will only remove the subscription when the unsubscribe method gets
	 * called.
	 * 
	 * @param subscriber Defines the listener which will get called when new messages arrive at the destination.
	 * @param durableSubscriberName Defines the name for a durable subscription.
	 * @throws JMSException Throws the exception when a subscription cannot be created due to a problem with a session or connection, or when the destination for the gateway is not a JMSTopic.
	 */
	public void registerDurableSubscriber(MessageListener subscriber,
			String durableSubscriberName) throws JMSException {
		if (destination instanceof Topic) {
			MessageConsumer messageConsumer = session.createDurableSubscriber(
					(Topic) destination, durableSubscriberName);
			messageConsumer.setMessageListener(subscriber);
			// If the user wants to unregister a durableSubscriber, it is
			// required that it is not active. All active ones will be
			// remembered.
			activeConsumers.put(durableSubscriberName, messageConsumer);
		} else {
			// logger.error("Cannot register a durable subscriber with a JMSQueue.");
			throw new JMSException(
					"Cannot register a durable subscriber with a Queue (i.e. destination is not a Topic).");
		}
	}

	/**
	 * Removes a durable subscription.  
	 * @param durableSubscriberName Defines the name for a durable subscription.
	 * @throws JMSException Throws an exception when the subscription cannot be removed due to a problem with a session or connection or the subscription does not exists.
	 */
	public void unregisterDurableSubsriber(String durableSubscriberName)
			throws JMSException {
		MessageConsumer consumer;
		// If it is an active consumer, close it and then remove it.
		if ((consumer = activeConsumers.remove(durableSubscriberName)) != null) {
			consumer.close();
		}
		session.unsubscribe(durableSubscriberName);
	}

	/**
	 * Method should be called when the gateway is no longer needed.
	 */
	public void close() {
		try {
			session.close();
			synchronized (SubscriberGateway.class) {
				if (--SubscriberGateway.count == 0) {
					connection.stop();
					connection.close();
					connection = null;
					// logger.debug("Last instance of ObserverGateway removed and connection closed.");
				}
			}
		} catch (JMSException e) {
			// logger.error(e.getMessage() + e.getStackTrace());
		} catch (NullPointerException e) {
			// logger.error(e.getMessage() + e.getStackTrace());
		}
	}

	/**
	 * @return The name of a destination associated with the gateway.
	 */
	public String getDestinationName() {
		return destinationName;
	}

}
